package com.fanghuaiming.tsec.pay.mqtransaction;

import com.alibaba.fastjson.JSONObject;
import com.fanghuaiming.tsec.common.model.enumeration.PayTransactionEnum;
import com.fanghuaiming.tsec.pay.dao.model.TsecPay;
import com.fanghuaiming.tsec.pay.service.PayService;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/****
 * @description: 分布式事务-支付事务监听器
 * @version:1.0.0
 * @author fanghuaiming
 * @data Created in 2020/12/1 下午4:40
 *
 */
@Slf4j
@Component
public class PayTransactionListener implements TransactionListener {

    /**
     * 共享以便回查
     */
    private ConcurrentHashMap<String, PayTransactionEnum> transactionMap = new ConcurrentHashMap<>();

    /**
     * 支付服务
     */
    @Autowired
    private PayService payService;

    /**
     * @Description: 执行本地事务
     *
     * @param:
     * @return:
     * @auther: fanghuaiming
     * @date: 2020/12/1 下午4:41
     * @version:1.0.0
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object object) {
        log.info("=========最终一致性-本地事务开始执行 事务id:{}=============",msg.getTransactionId());
        String transactionId = msg.getTransactionId();
        // 开始本地事务执行
        try {
            TsecPay tsecPay = (TsecPay)object;
            payService.updatePayStatus(tsecPay);
            //主动让消息进行回查
            if(tsecPay.getPayStatus() == 3){
                log.info("=========模拟网络超时,这里处理为unkown,Broker会主动进行回查=============");
                transactionMap.put(transactionId,PayTransactionEnum.UNKOWN);
                //模拟网络超时
                Thread.sleep(70000);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("执行支付操作出现出现异常:{}",e.getMessage(),e);
            log.error("=========本地事务出现异常,本地进行回滚 回滚消息,Broker端会删除半消息=============");
            transactionMap.put(transactionId,PayTransactionEnum.EXCEPTION);
            //回滚通知异常 回滚消息，Broker端会删除半消息
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        log.info("=========本地事务执行成功=============");
        //将状态存放到全局缓存中方便回查询--执行本地事务成功
        transactionMap.put(transactionId,PayTransactionEnum.SUCCESS);
        //正常提交 二次确认消息，然后消费者可以消费
        return LocalTransactionState.COMMIT_MESSAGE;
    }

    /**
     * @Description: 提供mq的broker进行回查
     *
     * @param:
     * @return:
     * @auther: fanghuaiming
     * @date: 2020/12/1 下午5:17
     * @version:1.0.0
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String transactionId = messageExt.getTransactionId();
        JSONObject jsonObject = JSONObject.parseObject(new String(messageExt.getBody()));
        //从消息中获取id和支付状态
        String uuid = jsonObject.getString("uuid");
        if(!uuid.isEmpty()){
            //这里可以去主动查询库的字段状态,也可以根据雪花算法id处理幂等性
        }
        log.info("=========事务回查开始执行 事务id:{}=============",transactionId);
        PayTransactionEnum payTransactionEnum = transactionMap.get(transactionId);
        if(payTransactionEnum.getCode() == 2){
            log.info("========= 回查结果 --> 回滚 =============");
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        if(payTransactionEnum.getCode() == 3){
            log.info("========= 回查结果 --> 未知 =============");
            return LocalTransactionState.UNKNOW;
        }
        log.info("========= 回查结果 --> 成功 =============");
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}
